// -*-mode:c++; coding:utf-8-*-

#ifndef _XBASE_TABLET_CHECKER_HPP_
#define _XBASE_TABLET_CHECKER_HPP_

#include <boost/bind.hpp>
#include <boost/asio.hpp>
#include <boost/noncopyable.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>

#include "global.hpp"
#include "table_manager.hpp"
#include "tablet_server_configuration.hpp"
#include "tablet.hpp"
#include "root_meta_tablet_client.hpp"

namespace xbase
{
	class tablet_checker
		: private boost::noncopyable
	{
	public:
		tablet_checker()
			: m_io_service()
			, m_timer(m_io_service)
			, m_duration(global::the_conf->tablet_check_interval) {
			set_timer();
		}

		void run() {
			LOG(INFO, "tablet_checker::run checker started");
			m_io_service.run();
		}

		void stop() {
			m_io_service.stop();
		}

	protected:
		template<typename Container>
		void list_online_tablet(Container &list) const {
			std::vector<table::pointer> table_list;
			global::the_table_manager->list_table(table_list);

			std::vector<tablet::pointer> tmp_list;
			BOOST_FOREACH(const table::pointer &elem, table_list)
			{
				elem->list_tablet(tmp_list);
				BOOST_FOREACH(const tablet::pointer &tablet, tmp_list)
				{
					if(tablet->is_online())
					{
						list.push_back(tablet);
					}
				}
				tmp_list.clear();
			}
		}
		
		void set_timer() {
			m_timer.expires_from_now(m_duration);
			m_timer.async_wait(boost::bind(&tablet_checker::check, this));
		}
		
		void check() {
			LOG(INFO, "tablet_checker::check check tablet info on root meta tablet");

			std::vector<tablet::pointer> tablet_list;
			list_online_tablet(tablet_list);

			// 没有子表，则不建立连接
			if(tablet_list.empty())
			{
				LOG(INFO, "tablet_checker::check got empty tablet list, done");
				set_timer();
				return;
			}

			const int retry_times = global::the_conf->retry_times;
			int retry_count = 0;
			while((! global::the_root_meta_manager->valid())
			      && (retry_count++ < retry_times))
			{
				LOG(ERROR, "tablet_checker::check root meta address is invalid, querying");
				global::the_service->query_root_meta_address();
				
				// 等待几秒钟，期望根子表可以更新
				boost::this_thread::sleep(boost::posix_time::seconds(global::the_conf->wait_seconds_for_master_reply));
			}
			if (! global::the_root_meta_manager->valid())
			{
				LOG(ERROR, "tablet_checker::check root meta address is still invalid after "
				    << retry_times << " retry(ies), abort checking");
				set_timer();
				return;
			}

			const protocol::server_address _addr = global::the_root_meta_manager->get();
			try {
				root_meta_tablet_client _client(_addr.m_hostname, _addr.m_port);
				if (! _client.is_open())
				{
					LOG(ERROR, "tablet_checker::check connect to root meta server failed, abort");
					global::the_root_meta_manager->set_invalid();
					global::the_service->query_root_meta_address(); // may be out of date
					set_timer();
					return;
				}

				protocol::tablet_info m_tablet_info;
				protocol::server_address m_address;
				uint64_t m_start_code;
				BOOST_FOREACH(tablet::pointer &tablet, tablet_list)
				{
					
					bool need_update = false;

					{
						if(tablet->is_root_meta_tablet()) // 跳过根子表，不检查
						{
							continue;
						}
						
						// 将该子表锁定，这将阻止主要紧缩和次要紧缩
						//Guard<ReadWriteLock::ReadLock> guard(tablet->lock().read_lock());
						boost::shared_lock<boost::shared_mutex> guard(tablet->lock());
						if(! tablet->is_online()) // double check, not online
						{
							LOG(DEBUG, "minor_compact_checker::check tablet "
							    << tablet->tablet_name()
							    << " is not online, skip it");
							continue;
						}
						
						// reset value
						m_tablet_info.m_tablet_id = INVALID_TABLET_ID;
						m_address.m_port = 0;
						m_start_code = 0;
						
						const bool ok = _client.get_tablet(m_tablet_info,
										   m_address,
										   m_start_code,
										   tablet->tablet_name());
						if(! ok) // 可能是网络原因或其它，通常会使其它子表也失败，所以直接退出这次检查
						{
							LOG(ERROR, "tablet_checker::check get tablet "
							    << tablet->tablet_name() << "'s info failed, abort checking");
							break;
						}
						
						if(m_tablet_info.m_tablet_id != tablet->tablet_id())
						{
							LOG(ERROR, "tablet_checker::check tablet id("
							    << m_tablet_info.m_tablet_id
							    << ") not matched to "
							    << tablet->tablet_id()
							    << ", skip it");
							continue;
						}
						
						// 首先检查子表服务器地址及起始码，若不匹配，则是该子表被重复分配，应该将子表下线，
						// 且不能更改根子表中的信息
						if((m_address.m_port != global::the_address_info->m_port)
						   || (m_address.m_startcode != global::the_address_info->m_startcode)
						   || (m_address.m_hostname != global::the_address_info->m_hostname))
						{
							LOG(WARN, "tablet_checker::check tablet "
							    << tablet->tablet_name()
							    << "'s address in root meta tablet is "
							    << m_address.to_string()
							    << ", so remove this tablet from this tablet server");
						}
						else
						{
							// 检查子表状态：首先，不可能是分裂状态，因为一旦分裂就会将它下线；
							assert(! m_tablet_info.status_is_split());
						
							// 如果是下线状态，则一定出问题了，就将子表下线，且不能更改根子表中的信息
							if(! m_tablet_info.status_is_online())
							{
								LOG(WARN, "tablet_checker::check tablet "
								    << tablet->tablet_name()
								    << "'s status in root meta tablet is offline,"
								    << " so remove this tablet from this tablet server");
							}
							// 如果是删除状态，则是正常情况，即该表被删除了，此时应该将子表下线，并置根子表中的状态为下线状态，
							// 以使master可以删除该子表
							else if(m_tablet_info.status_is_deleted())
							{
								LOG(INFO, "tablet_checker::check tablet "
								    << tablet->tablet_name()
								    << "'s status in root meta tablet is "
								    << "deleted"
								    << ", so unload this tablet");
								need_update = true;
							}
							else // 正常，下一个
							{
								continue;
							}
						}

						// 这个时候，影响到读写操作，要对子表加写锁才能进行!
						// 因此，在获取读锁的情况下，将状态设为下线，这会阻止读写操作及紧缩操作
						// 然后，再释放读锁，加上写锁
						tablet->set_online(false);
					}
					
					// 下线，根据情况修改根子表中的数据
					{
						// 禁止任何对该子表的操作
						//Guard<ReadWriteLock::WriteLock> guard(tablet->lock().write_lock());
						boost::unique_lock<boost::shared_mutex> guard(tablet->lock());
						tablet->set_online(true); // 将状态设为上线，因为若删除失败的话，只有这样才能使它下次还被检查并再试一次删除
						
						tablet->do_minor_compact_forced(); // 确保保存了所有数据，并清空了日志
						
						table::pointer table = global::the_table_manager->get_table(tablet->table_name());
						if(! table)
						{
							LOG(FATAL, "tablet_checker::check get table "
							    << tablet->table_name() << "  failed");
							
							global::the_service->report_tablet_server_quit();
							assert(false); // 不可能的事情，否则不可恢复！
						}
						const std::size_t num = table->remove_tablet(tablet);
						LOG_IF_ELSE((num != 1),
							    WARN,
							    "tablet_checker::check remove tablet "
							    << tablet->tablet_name() << " return " << num,
							    INFO,
							    "tablet_checker::check removed tablet "
							    << tablet->tablet_name()
							    << " from this tablet server");
						// 重要！
						// 删除成功后，再将子表状态设为下线。因为这时候可能正有读写请求或紧缩操作在等待该锁，
						// 而在此时释放后，只有状态为下线，才能使那些操作失败，不再进行。
						tablet->set_online(false);
					} // 释放该子表的锁
					if(need_update) // 更新根子表中的状态为下线
					{
						try {
							const std::string log_message = "tablet_server: tablet status in root meta tablet"
								" is deleted, so unload this tablet and set it to be offline";
							m_tablet_info.status_clear_online();
							const bool ok = _client.update_tablet(tablet->tablet_name(),
											      m_tablet_info,
											      log_message);
							if(! ok) // 更新失败，则说明与根子表的连接有问题，退出子表服务器
							{
								LOG(FATAL, "tablet_checker::check updating tablet status failed");
								global::the_service->report_tablet_server_quit();
								assert(false);
							}
						}
						catch(const std::exception &e) // 更新失败，则说明与根子表的连接有问题，退出子表服务器
						{
							LOG(FATAL, "tablet_checker::check updating tablet status failed: " << e.what());
							global::the_service->report_tablet_server_quit();
							assert(false);
						}
					}
				}
			}
			catch (const std::exception &e)
			{
				LOG(ERROR, "tablet_checker::check got error message: " << e.what());
				global::the_root_meta_manager->set_invalid();
				global::the_service->query_root_meta_address(); // may be out of date
			}

			set_timer();
			LOG(INFO, "tablet_checker::check checking done");
		}

	private:
		boost::asio::io_service m_io_service;
		boost::asio::deadline_timer m_timer;
		boost::posix_time::seconds m_duration;
	};

} // namespace xbase

#endif	// _XBASE_TABLET_CHECKER_HPP_
